package com.zhang.third.day04;

import com.zhang.third.utils.ClickEventSource;
import com.zhang.third.utils.Event;
import com.zhang.third.utils.UserViewCountPerWindow;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @title: 使用KeyedProcessFunction实现Example4的功能
 * @author: zhang
 * @date: 2022/4/7 15:58
 * process 内部只维护累加器
 */
public class Example6 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickEventSource())
                .keyBy(r -> r.key)
                .process(new MyTumblingProcessingTimeWindow(5000L))
                .print();

        env.execute();
    }

    public static class MyTumblingProcessingTimeWindow extends KeyedProcessFunction<String, Event, UserViewCountPerWindow> {
        private long windowSize;

        // mapstate
        // key:窗口开始时间
        // value:属于该窗口的所有元素集合
        private MapState<Long, Long> mapState;

        public MyTumblingProcessingTimeWindow(long windowSize) {
            this.windowSize = windowSize;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            mapState = getRuntimeContext().getMapState(
                    new MapStateDescriptor<Long, Long>(
                            "map-state",
                            Types.LONG,
                            Types.LONG
                    )
            );
        }

        @Override
        public void processElement(Event value, KeyedProcessFunction<String, Event, UserViewCountPerWindow>.Context ctx, Collector<UserViewCountPerWindow> out) throws Exception {
            // 为事件分配窗口
            // 计算窗口开始事件
            long currTs = ctx.timerService().currentProcessingTime();
            long windowStartTime = currTs - currTs % windowSize;
            // 实现的是Example4中的.window()和增量聚合函数方法
            if (!mapState.contains(windowStartTime)) {
                mapState.put(windowStartTime, 1L);
            } else {
                mapState.put(windowStartTime, mapState.get(windowStartTime) + 1L);
            }
            // 注册窗口结束时间-1ms的定时器
            ctx.timerService().registerProcessingTimeTimer(windowStartTime + windowSize - 1L);
        }

        // onTimer实现的是Example3中的process方法的逻辑
        @Override
        public void onTimer(long timestamp, KeyedProcessFunction<String, Event, UserViewCountPerWindow>.OnTimerContext ctx, Collector<UserViewCountPerWindow> out) throws Exception {
            long windowStartTime = timestamp + 1L - windowSize;
            long windowEndTime = windowStartTime + windowSize;
            String username = ctx.getCurrentKey();
            long count = mapState.get(windowStartTime);
            out.collect(new UserViewCountPerWindow(
                    username,
                    count,
                    windowStartTime,
                    windowEndTime
            ));

            // 销毁窗口
            mapState.remove(windowStartTime);
        }
    }
}
